package cn.ztelovecly.springboot.framework.message;

import cn.ztelovecly.springboot.framework.util.SettingUtil;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.aliyun.openservices.ons.api.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

@Component
public class BaseCoumser implements ApplicationContextAware {
    // 获取所有的XNConsumerMessage 子类
    private ApplicationContext applicationContext;

    protected Logger logger = LoggerFactory.getLogger(this.getClass());

    @PostConstruct
    public void init() {
        String deployMode = SettingUtil.getProperty("", "deploy.mode").toLowerCase();
        String mqType = SettingUtil.getProperty("","mq.type");
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.AccessKey, SettingUtil.getProperty("", "aliyun.accesskey.id"));
        properties.put(PropertyKeyConst.SecretKey, SettingUtil.getProperty("", "aliyun.accesskey.secret"));
        properties.put(PropertyKeyConst.ConsumerId, SettingUtil.getProperty("", "mq.consumer.id"));
        properties.put(PropertyKeyConst.ConsumeThreadNums, 1);

        if(mqType!=null && !mqType.equals("")){
            switch (mqType.toLowerCase()){
                case "ons" :
                    ConsumerforONS(properties);
                    break;
                case "rocketmq" :
                    ConsumerforRocketMQ(properties);
                    break;
                default:
                    break;
            }
        }
        // 兼容老版本的设置方式
        else {
            if (!"production".equals(deployMode)) {
                ConsumerforRocketMQ(properties);
            } else {
                ConsumerforONS(properties);
            }
        }
    }

    private void ConsumerforRocketMQ(Properties properties) {
        DefaultMQPushConsumer consumer =
                new DefaultMQPushConsumer(properties.getProperty(PropertyKeyConst.ConsumerId));
        consumer.setNamesrvAddr(SettingUtil.getProperty("", "local.mq.address"));
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        try {
            Map<String, ConsumerMessage> beanMaps = this.getAllObjectByType(ConsumerMessage.class);
            final Map<String, ConsumerMessage> mymessages = new HashMap<>();
            if (beanMaps != null && !beanMaps.isEmpty()) {
                for (String beanName : beanMaps.keySet()) {
                    ConsumerMessage message = beanMaps.get(beanName);
                    mymessages.put(message.getTopic() + "##" + message.getTag(), message);
                    //循环监听消息
                    consumer.subscribe(message.getTopic(),"*");
                }
                consumer.registerMessageListener(
                        new MessageListenerConcurrently() {
                            @Override
                            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext Context) {
                                Message msg = list.get(0);
                                if (mymessages.get(msg.getTopic() + "##" + msg.getTags()) != null) {
                                    ConsumerMessage consumerMessage = mymessages.get(msg.getTopic() + "##" + msg.getTags());
                                    return consumerMessage.consumerMessage(new String(msg.getBody()));
                                }
                                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                            }
                        });
            } else {
                logger.debug("暂未找到消费消息...");
//                System.out.println("暂未找到消费消息...");
            }
            consumer.start();
            logger.debug("MQ Consumer Started...");
//            System.out.println("MQ Consumer Started...");
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
//            e.printStackTrace();
        }
    }

    /**
     * ONS使用
     * @param properties
     */
    private void ConsumerforONS(Properties properties) {
        Consumer consumer = ONSFactory.createConsumer(properties);
        Map<String, ConsumerMessage> beanMaps = this.getAllObjectByType(ConsumerMessage.class);
        final Map<String, ConsumerMessage> mymessages = new HashMap<>();
        if (beanMaps != null && !beanMaps.isEmpty()) {
            for (String beanName : beanMaps.keySet()) {
                ConsumerMessage message = beanMaps.get(beanName);
                mymessages.put(message.getTopic() + "##" + message.getTag(), message);
                //循环监听消息
                logger.debug("ONS parse topic : "+message.getTopic()+" tag : "+message.getTag());
//                System.out.println("ONS parse topic : "+message.getTopic()+" tag : "+message.getTag());
                consumer.subscribe(message.getTopic(),"*",new MessageListener() {
                    @Override
                    public Action consume(com.aliyun.openservices.ons.api.Message message, ConsumeContext consumeContext) {
                        if (mymessages.get(message.getTopic() + "##" + message.getTag()) != null) {
                            ConsumerMessage consumerMessage = mymessages.get(message.getTopic() + "##" + message.getTag());
                            ConsumeConcurrentlyStatus status = consumerMessage.consumerMessage(new String(message.getBody()));
                            if(status.equals(ConsumeConcurrentlyStatus.CONSUME_SUCCESS)){
                                return  Action.CommitMessage;
                            }else {
                                return Action.ReconsumeLater;
                            }
                        }
                        return Action.ReconsumeLater;

                    }
                });
            }
        } else {
            logger.debug("暂未找到消费消息...");
//            System.out.println("暂未找到消费消息...");
        }
        consumer.start();
        logger.debug("ONS Consumer Started...");
//        System.out.println("ONS Consumer Started...");
    }
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    public <T> Map<String,T> getAllObjectByType(Class<T> clazz){
        return applicationContext.getBeansOfType(clazz);
    }
}
